package com.lxl.core;

import com.lxl.config.TaskStatus;
import com.lxl.pojo.InterfaceMonitor;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobListener;
import org.pentaho.di.job.JobMeta;

import java.util.Iterator;
import java.util.Map;

public class EtlJobTask extends BaseTask {

    public TaskResultMeta runTask() throws KettleException {
        TaskResultMeta resultMeta = new TaskResultMeta();

        EtlJobListener jobListener = new EtlJobListener();
        JobMeta jobMeta = new JobMeta(taskDataMeta.taskLocalConfigFilePath, null);
        Job job = new Job(null,jobMeta);

        jobListener.setInterfaceMonitor(this.taskDataMeta.interfaceMonitor);
        job.addJobListener(jobListener);

        Iterator<Map.Entry<String, String>> jobParam = taskDataMeta.etlInputVariables.entrySet().iterator();
        while (jobParam.hasNext()){
            Map.Entry<String, String> entry = jobParam.next();
            job.setVariable(entry.getKey(), entry.getValue());
        }

        job.start();
        job.waitUntilFinished();
        
        resultMeta.setCode(job.getErrors());
        if (resultMeta.getCode() > 0){
            resultMeta.setResult(null);
        }else{
            resultMeta.setResult(job.getVariable(taskDataMeta.etlResultKey));
        }

        return resultMeta;
    }

    class EtlJobListener implements JobListener{
        private InterfaceMonitor interfaceMonitor;
        private long start,end;
        private void setInterfaceMonitor(InterfaceMonitor interfaceMonitor){
            this.interfaceMonitor = interfaceMonitor;
        }
        private InterfaceMonitor getInterfaceMonitor(){
            return interfaceMonitor;
        }

        @Override
        public void jobStarted(Job job) throws KettleException {
            start = System.currentTimeMillis();
            interfaceMonitor.setStatus(TaskStatus.START);
        }

        @Override
        public void jobFinished(Job job) throws KettleException {
            end = System.currentTimeMillis();
            interfaceMonitor.setStatus(TaskStatus.SUCCESS);
            interfaceMonitor.setTimeConsuming(end-start);
        }
    }
}
